package com.shujia.core

import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.Logger

import scala.util.Random

object Demo19SparkPI extends Logging {

  val logger: Logger = log

  def main(args: Array[String]): Unit = {
    // 随机生成点

    //    for (elem <- 1 to 100) {
    //      println((Random.nextDouble() * 2 - 1, Random.nextDouble() * 2 - 1))
    //    }

    /**
     * 从args接受一个参数 表示生成的分区的数量
     */
    if (args.isEmpty) {
      logger.error("请指定分区数量！")
      return
    }

    val numSlices: Int = args.head.toInt


    // 给每个分区生成10w个点
    val pointNumPerPartition: Int = 100000

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo19SparkPI")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val sourceRDD: RDD[Int] = sc.parallelize(1 to pointNumPerPartition * numSlices, numSlices)

    // 获取RDD的分区数
    logger.info(sourceRDD.getNumPartitions.toString)

    // 在每个分区 基于它的数据条数 生成对应N个点
    val sumPI: Double = sourceRDD.map(line => {
      (Random.nextDouble() * 2 - 1, Random.nextDouble() * 2 - 1)
    })
      // 在每个分区内计算出一个π值，最终10个取平均
      .mapPartitionsWithIndex((index, iter) => {
        // 让每个分区内的点 加上 分区编号
        logger.debug(s"当前的分区为：$index")
        iter.map(t2 => (t2._1, t2._2, index))
      })
      // 按照分区编号进行分组，即可实现对同一个分区内的点进行处理 计算π值
      .groupBy(_._3)
      .map(kv => {
        val index: Int = kv._1
        val pointNum: Int = kv._2.size

        /**
         * Iterable提供了一个count方法，可以给定一个规则去统计符合该规则的数据条数
         * 类似 .filter(t3 => (t3._1 * t3._1 + t3._2 * t3._2) <= 1).size
         */
        val pointNumInCircle: Int = kv._2.count(t3 => (t3._1 * t3._1 + t3._2 * t3._2) <= 1)
        val PI: Double = 4.0 * pointNumInCircle / pointNum
        (index, PI)
      })
      // 计算所有分区内PI值的平均值
      .map(_._2)
      .reduce(_ + _)

    // 平均的PI值
    val avgPI: Double = sumPI / numSlices
    logger.info(avgPI.toString)
    println(avgPI.toString)

    while (true) {

    }


  }

}
